package com.heiye.user.biz.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.RandomUtil;
import cn.hutool.json.JSONUtil;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.heiye.count.api.dto.FindUserCountsByIdRspDTO;
import com.heiye.framework.common.enums.DeletedEnum;
import com.heiye.framework.common.enums.StatusEnum;
import com.heiye.framework.common.exception.BizException;
import com.heiye.framework.common.response.Response;
import com.heiye.framework.common.util.JsonUtils;
import com.heiye.framework.common.util.NumberUtils;
import com.heiye.framework.common.util.ParamUtils;
import com.heiye.user.api.dto.req.FindUserByIdReqDTO;
import com.heiye.user.api.dto.req.FindUsersByIdsReqDTO;
import com.heiye.user.api.dto.resp.RegisterUserRspDTO;
import com.heiye.user.api.dto.resp.UserLanguageRspDTO;
import com.heiye.user.biz.constant.MQConstants;
import com.heiye.user.biz.constant.RedisKeyConstants;
import com.heiye.user.biz.constant.RoleConstants;
import com.heiye.user.biz.domain.dataobject.*;
import com.heiye.user.biz.domain.mapper.*;
import com.heiye.user.api.dto.req.RegisterUserReqDTO;
import com.heiye.user.biz.enums.LanguageTypeEnum;
import com.heiye.user.biz.enums.RegisterTypeEnum;
import com.heiye.user.biz.enums.ResponseCodeEnum;
import com.heiye.user.api.dto.resp.FindUserByIdRspDTO;
import com.heiye.user.biz.model.dto.resp.FindUserRspDTO;
import com.heiye.user.biz.model.vo.*;
import com.heiye.user.biz.rpc.CountRpcService;
import com.heiye.user.biz.rpc.DistributedIdGeneratorRpcService;
import com.heiye.user.biz.rpc.OssRpcService;
import com.heiye.user.biz.service.UserService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.heiye.framework.biz.context.thread.LoginUserContextHolder;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionTemplate;

import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author: heiye
 * @date: 2024/09/29 下午3:07
 * @version: v1.0.0
 * @description: 用户ServerImpl类
 */
@Slf4j
@Service
public class UserServiceImpl implements UserService {
    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private UserDOMapper userDOMapper;

    @Resource
    private UserRoleDOMapper userRoleDOMapper;

    @Resource
    private RoleDOMapper roleDOMapper;

    @Resource
    private UserLanguageDOMapper userLanguageDOMapper;

    @Resource
    private TransactionTemplate transactionTemplate;

    @Resource
    private LanguageDOMapper languageDOMapper;

    @Resource
    private OssRpcService ossRpcService;

    @Resource(name = "taskExecutor")
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Resource
    private DistributedIdGeneratorRpcService distributedIdGeneratorRpcService;

    @Resource
    private CountRpcService countRpcService;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 用户信息本地缓存
     */
    private static final Cache<Long, FindUserByIdRspDTO> LOCAL_CACHE = Caffeine.newBuilder()
            // 设置初始容量为 10000 个条目
            .initialCapacity(10000)
            // 设置缓存的最大容量为 10000 条
            .maximumSize(10000)
            // 设置缓存条目在写入后 1 小时过期
            .expireAfterWrite(1, TimeUnit.HOURS)
            .build();

    /**
     * 用户主页信息本地缓存
     */
    private static final Cache<Long, FindUserProfileRspVO> PROFILE_LOCAL_CACHE = Caffeine.newBuilder()
            // 设置初始容量为 10000 个条目
            .initialCapacity(10000)
            // 设置缓存的最大容量为 10000 条
            .maximumSize(10000)
            // 设置缓存条目在写入后 5 分钟过期
            .expireAfterWrite(5, TimeUnit.MINUTES)
            .build();

    /**
     * 用户注册
     *
     * @param registerUserReqDTO
     * @return
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public Response<RegisterUserRspDTO> register(RegisterUserReqDTO registerUserReqDTO) {
        // 获取用户标识和注册类型
        String identifier = registerUserReqDTO.getIdentifier();
        Integer type = registerUserReqDTO.getType();
        // 创建注册枚举
        RegisterTypeEnum registerTypeEnum = RegisterTypeEnum.valueOf(type);

        // 创建查询数据库得到的 UserDO
        UserDO userDO;

        switch (registerTypeEnum) {
            // 通过手机号查找用户否已经注册
            case PHONE -> userDO = userDOMapper.selectByPhone(identifier);
            // 通过邮箱查找用户否已经注册
            case EMAIL -> userDO = userDOMapper.selectByEmail(identifier);
            default -> throw new BizException(ResponseCodeEnum.REGISTER_TYPE_ERROR);
        }

        log.info("==> 用户是否注册, mobile: {}, user: {}", identifier, JSONUtil.toJsonStr(userDO));

        // 若已注册，直接返回用户id
        if (Objects.nonNull(userDO)) {
            // 获取用户的语言信息
            List<UserLanguageDO> userLanguageDOS = userLanguageDOMapper.selectByUserId(userDO.getId());
            // 构建返回数据
            RegisterUserRspDTO registerUserRspDTO = RegisterUserRspDTO.builder()
                    .id(userDO.getId())
                    .isRegister(CollUtil.isNotEmpty(userLanguageDOS))
                    .build();

            return Response.success(registerUserRspDTO);
        }

        // 获取redis里用户的全局自增id
//        Long uid = redisTemplate.opsForValue().increment(RedisKeyConstants.INTERACT_ID_GENERATOR_KEY);

        // RPC: 调用分布式 ID 生成服务生成 InteractID
        String interactId = distributedIdGeneratorRpcService.getInteractId();

        // RPC: 调用分布式 ID 生成服务生成用户 ID
        String userIdStr = distributedIdGeneratorRpcService.getUserId();
        Long userId = Long.valueOf(userIdStr);

        // 初始化用户信息
        userDO = UserDO.builder()
                .id(userId)
                .uid(String.valueOf(interactId))
                .displayName("用户_" + interactId)
                .photoUrl("")
                .synopsis("")
                .phoneNumber("")
                .email("")
                .password("")
                .createTime(LocalDateTime.now())
                .updateTime(LocalDateTime.now())
                .lastSignedInTime(LocalDateTime.now())
                .status(StatusEnum.ENABLE.getValue())
                .isDeleted(DeletedEnum.NO.getValue())
                .build();

        // 配置用户标识
        switch (registerTypeEnum) {
            case PHONE -> userDO.setPhoneNumber(identifier);
            case EMAIL -> userDO.setEmail(identifier);
        }

        // 添加用户信息
        userDOMapper.insert(userDO);

        // 获取刚刚添加入库的用户 ID
//        Long userId = userDO.getId();

        log.info("=====> 用户ID {}", userId);

        // 构建用户权限关联信息
        UserRoleDO userRole = UserRoleDO.builder()
                .userId(userId)
                .roleId(RoleConstants.COMMON_USER_ROLE_ID)
                .createTime(LocalDateTime.now())
                .updateTime(LocalDateTime.now())
                .isDeleted(DeletedEnum.NO.getValue())
                .build();

        // 添加用户权限关联信息
        userRoleDOMapper.insert(userRole);

        // 查询
        RoleDO roleDO = roleDOMapper.selectByPrimaryKey(RoleConstants.COMMON_USER_ROLE_ID);

        // 将该用户的角色 ID 存入 Redis 中，指定初始容量为 1，这样可以减少在扩容时的性能开销
        List<String> roles = new ArrayList<>(1);

        // 添加角色权限
        roles.add(roleDO.getRoleKey());

        // 构建key
        String userRolesKey = RedisKeyConstants.buildUserRoleKey(userId);

        // 放入到redis中
        redisTemplate.opsForValue().set(userRolesKey, JSONUtil.toJsonStr(roles));

        // 构建返回数据
        RegisterUserRspDTO registerUserRspDTO = RegisterUserRspDTO.builder()
                .id(userDO.getId())
                .isRegister(false)
                .build();

        return Response.success(registerUserRspDTO);
    }

    /**
     * 修改用户的母语，以及感兴趣的语言
     *
     * @param userLanguageUpdateReqVO
     * @return
     */
    @Override
    public Response<?> userLanguageUpdate(UserLanguageUpdateReqVO userLanguageUpdateReqVO) {
        return transactionTemplate.execute(status -> {
                    try {
                        // 获取该用户id
                        Long userId = LoginUserContextHolder.getUserId();

                        // 先删除掉用户的之前保留的语言信息
                        userLanguageDOMapper.deleteById(userId);

                        // 处理用户母语列表
                        List<UserLanguageReqVO> nativeLanguage = userLanguageUpdateReqVO.getNativeLanguage();

                        // 构建 数据库插入数据
                        List<UserLanguageDO> userLanguageDOS = Lists.newArrayList();

                        // 循环
                        for (UserLanguageReqVO userLanguageReqVO : nativeLanguage) {

                            // 构建插入数据
                            UserLanguageDO userLanguageDO = UserLanguageDO.builder()
                                    .userId(userId)
                                    .languageId(userLanguageReqVO.getId())
                                    .level(userLanguageReqVO.getLevel())
                                    .type(LanguageTypeEnum.NATIVE.getValue())
                                    .createTime(LocalDateTime.now())
                                    .updateTime(LocalDateTime.now())
                                    .build();

                            userLanguageDOS.add(userLanguageDO);

                        }

                        // 处理用户感兴趣语言列表
                        List<UserLanguageReqVO> interestLanguage = userLanguageUpdateReqVO.getInterestLanguage();

                        // 循环
                        for (UserLanguageReqVO userLanguageReqVO : interestLanguage) {

                            // 构建插入数据
                            UserLanguageDO userLanguageDO = UserLanguageDO.builder()
                                    .userId(userId)
                                    .languageId(userLanguageReqVO.getId())
                                    .level(userLanguageReqVO.getLevel())
                                    .type(LanguageTypeEnum.INTEREST.getValue())
                                    .createTime(LocalDateTime.now())
                                    .updateTime(LocalDateTime.now())
                                    .build();

                            userLanguageDOS.add(userLanguageDO);

                        }

                        // 插入用户母语列表数据
                        userLanguageDOMapper.insert(userLanguageDOS);


                        // 异步将存入 Redis 缓存中的用户信息删除，提升响应速度
                        threadPoolTaskExecutor.submit(() -> {
                            // 构建用户信息 Redis KEY
                            String userInfoRedisKey = RedisKeyConstants.buildUserInfoKey(userId);
                            redisTemplate.delete(userInfoRedisKey);
                        });

                        return Response.success();

                    } catch (Exception e) {
                        // 标记事务为回滚
                        status.setRollbackOnly();
                        log.error("==> 系统添加用户语言信息异常: ", e);
                        throw new BizException(ResponseCodeEnum.USER_LANGUAGE_INT_FAIL);
                    }
                }
        );
    }

    /**
     * 查找用户基本信息
     *
     * @return
     */
    @Override
    public Response<FindUserRspDTO> find() {
        // 获取用户ID
        Long userId = LoginUserContextHolder.getUserId();

        // 查询用户信息
        UserDO userDO = userDOMapper.selectById(userId);

        // 查询用户语言信息
        List<UserLanguageDO> userLanguageDOS = userLanguageDOMapper.selectByUserId(userId);

        // 判空
        if (CollUtil.isEmpty(userLanguageDOS)) {
            throw new BizException(ResponseCodeEnum.USER_LANGUAGE_NOT_FOUND);
        }

        // 使用 stream 流转化成 languageIds 数组
        List<Long> languageIds = userLanguageDOS.stream()
                .map(UserLanguageDO::getLanguageId)
                .toList();

        // 通过 languageIds 查询出对应的语言数据
        List<LanguageDO> languageDOS = languageDOMapper.selectByLanguageIds(languageIds);

        // 构建出 languageDOMap 以 id 为 key, name 为 value
        Map<Long, LanguageDO> languageDOMap = languageDOS.stream()
                .collect(Collectors.toMap(LanguageDO::getId, value -> value));

        // 构建用户语言信息反参
        List<UserLanguageRspDTO> nativeLanguage = Lists.newArrayList();
        List<UserLanguageRspDTO> interestLanguage = Lists.newArrayList();

        // 遍历用户语言信息
        for (UserLanguageDO userLanguageDO : userLanguageDOS) {
            // 获取语言信息
            LanguageDO languageDO = languageDOMap.get(userLanguageDO.getLanguageId());
            // 构建反参
            UserLanguageRspDTO userLanguageRspDTO = UserLanguageRspDTO.builder()
                    .id(userLanguageDO.getLanguageId())
                    .name(languageDO.getName())
                    .initialsName(languageDO.getInitialsName())
                    .level(userLanguageDO.getLevel())
                    .build();

            LanguageTypeEnum languageTypeEnum = LanguageTypeEnum.valueOf(userLanguageDO.getType());

            // 判断加入的语言数组
            switch (languageTypeEnum) {
                // 添加到母语中
                case NATIVE -> nativeLanguage.add(userLanguageRspDTO);
                // 添加到感兴趣语言中
                case INTEREST -> interestLanguage.add(userLanguageRspDTO);
            }

        }

        // 构建反参
        FindUserRspDTO findUserRspDTO = FindUserRspDTO.builder()
                .id(userDO.getId())
                .displayName(userDO.getDisplayName())
                .photoUrl(userDO.getPhotoUrl())
                .synopsis(userDO.getSynopsis())
                .nativeLanguage(nativeLanguage)
                .interestLanguage(interestLanguage)
                .build();

        return Response.success(findUserRspDTO);
    }

    /**
     * 更新用户信息
     *
     * @param updateUserInfoReqVO
     * @return
     */
    @Override
    public Response<?> updateUserInfo(UpdateUserInfoReqVO updateUserInfoReqVO) {
        // 被更新的用户 ID
        Long userId = updateUserInfoReqVO.getUserId();
        // 当前登录的用户 ID
        Long loginUserId = LoginUserContextHolder.getUserId();

        // 非号主本人，无法修改其个人信息
        if (Objects.equals(loginUserId, userId)) {
            throw new BizException(ResponseCodeEnum.CANT_UPDATE_OTHER_USER_PROFILE);
        }

        UserDO userDO = new UserDO();

        // 设置当前需要更新的用户ID
        userDO.setId(userId);
        // 标识位：是否需要更新
        boolean needUpdate = false;

        // 昵称
        String displayName = updateUserInfoReqVO.getDisplayName();
        if (StringUtils.isNotBlank(displayName)) {
            Preconditions.checkArgument(ParamUtils.checkDisplayName(displayName), ResponseCodeEnum.DISPLAY_NAME_VALID_FAIL.getErrorMessage());
            userDO.setDisplayName(displayName);
            needUpdate = true;
        }

        // 个人简介
        String synopsis = updateUserInfoReqVO.getSynopsis();
        if (StringUtils.isNotBlank(synopsis)) {
            Preconditions.checkArgument(ParamUtils.checkLength(synopsis, 100), ResponseCodeEnum.SYNOPSIS_VALID_FAIL.getErrorMessage());
            userDO.setSynopsis(synopsis);
            needUpdate = true;
        }

        if (needUpdate) {
            // 删除用户缓存
            deleteUserRedisCache(userId);

            // 更新用户信息
            userDO.setUpdateTime(LocalDateTime.now());
            userDOMapper.updateByPrimaryKeySelective(userDO);

            // 延时双删
            sendDelayDeleteUserRedisCacheMQ(userId);
        }

        return Response.success();
    }

    /**
     * 根据用户 ID 查询用户信息
     *
     * @param findUserByIdReqDTO
     * @return
     */
    @Override
    public Response<FindUserByIdRspDTO> findById(FindUserByIdReqDTO findUserByIdReqDTO) {
        // 用户 ID
        Long userId = findUserByIdReqDTO.getId();

        // 先从本地缓存中查询
        FindUserByIdRspDTO findUserByIdRspDTOLocalCache = LOCAL_CACHE.getIfPresent(userId);
        if (Objects.nonNull(findUserByIdRspDTOLocalCache)) {
            log.info("==> 命中了本地缓存；{}", findUserByIdRspDTOLocalCache);
            return Response.success(findUserByIdRspDTOLocalCache);
        }

        // 用户缓存 Redis KEY
        String userInfoRedisKey = RedisKeyConstants.buildUserInfoKey(userId);

        // 先从 Redis 缓存中查询
        String userInfoRedisValue = (String) redisTemplate.opsForValue().get(userInfoRedisKey);

        // 若 Redis 缓存中存在该用户信息
        if (StringUtils.isNotBlank(userInfoRedisValue)) {
            // 将存储的 Json 字符串转换成对象，并返回
            FindUserByIdRspDTO findUserByIdRepDTO = JsonUtils.parseObject(userInfoRedisValue, FindUserByIdRspDTO.class);
            // 异步线程中将用户信息存入本地缓存
            threadPoolTaskExecutor.submit(() -> {
                if (Objects.nonNull(findUserByIdRepDTO)) {
                    // 将用户信息写入本地缓存
                    LOCAL_CACHE.put(userId, findUserByIdRepDTO);
                }
            });
            return Response.success(findUserByIdRepDTO);
        }

        // 否则, 从数据库中查询
        // 根据用户 ID 查询用户信息
        UserDO userDO = userDOMapper.selectById(userId);

        // 判空
        if (Objects.isNull(userDO)) {
            threadPoolTaskExecutor.execute(() -> {
                // 防止缓存穿透，将空数据存入 Redis 缓存 (过期时间不宜设置过长)
                // 保底1分钟 + 随机秒数
                long expireSeconds = 60 + RandomUtil.randomInt(60);
                redisTemplate.opsForValue().set(userInfoRedisKey, "null", expireSeconds, TimeUnit.SECONDS);
            });
            throw new BizException(ResponseCodeEnum.USER_NOT_FOUND);
        }

        // 查询用户语言信息
        List<UserLanguageDO> userLanguageDOS = userLanguageDOMapper.selectByUserId(userId);

        // 判空
        if (CollUtil.isEmpty(userLanguageDOS)) {
            throw new BizException(ResponseCodeEnum.USER_LANGUAGE_NOT_FOUND);
        }

        // 使用 stream 流转化成 languageIds 数组
        List<Long> languageIds = userLanguageDOS.stream()
                .map(UserLanguageDO::getLanguageId)
                .toList();

        // 通过 languageIds 查询出对应的语言数据
        List<LanguageDO> languageDOS = languageDOMapper.selectByLanguageIds(languageIds);

        // 构建出 languageDOMap 以 id 为 key, name 为 value
        Map<Long, LanguageDO> languageDOMap = languageDOS.stream()
                .collect(Collectors.toMap(LanguageDO::getId, value -> value));

        // 构建用户语言信息反参
        List<UserLanguageRspDTO> nativeLanguage = Lists.newArrayList();
        List<UserLanguageRspDTO> interestLanguage = Lists.newArrayList();

        // 遍历用户语言信息
        for (UserLanguageDO userLanguageDO : userLanguageDOS) {
            // 获取语言信息
            LanguageDO languageDO = languageDOMap.get(userLanguageDO.getLanguageId());
            // 构建反参
            UserLanguageRspDTO userLanguageRspDTO = UserLanguageRspDTO.builder()
                    .id(userLanguageDO.getLanguageId())
                    .name(languageDO.getName())
                    .initialsName(languageDO.getInitialsName())
                    .level(userLanguageDO.getLevel())
                    .build();

            LanguageTypeEnum languageTypeEnum = LanguageTypeEnum.valueOf(userLanguageDO.getType());

            // 判断加入的语言数组
            switch (languageTypeEnum) {
                // 添加到母语中
                case NATIVE -> nativeLanguage.add(userLanguageRspDTO);
                // 添加到感兴趣语言中
                case INTEREST -> interestLanguage.add(userLanguageRspDTO);
            }
        }

        // 构建反参 DTO
        FindUserByIdRspDTO findUserByIdRepDTO = FindUserByIdRspDTO.builder()
                .id(userDO.getId())
                .displayName(userDO.getDisplayName())
                .photoUrl(userDO.getPhotoUrl())
                .nativeLanguage(nativeLanguage)
                .interestLanguage(interestLanguage)
                .build();

        // 异步将用户信息存入 Redis 缓存，提升响应速度
        threadPoolTaskExecutor.submit(() -> {
            // 过期时间（保底1天 + 随机秒数，将缓存过期时间打散，防止同一时间大量缓存失效，导致数据库压力太大）
            long expireSeconds = 60 * 60 * 24 + RandomUtil.randomInt(60 * 60 * 24);

            redisTemplate.opsForValue().set(userInfoRedisKey, JsonUtils.toJsonString(findUserByIdRepDTO), expireSeconds, TimeUnit.SECONDS);
        });

        return Response.success(findUserByIdRepDTO);
    }

    /**
     * 批量根据用户 ID 查询用户信息
     *
     * @param findUsersByIdsReqDTO
     * @return
     */
    @Override
    public Response<List<FindUserByIdRspDTO>> findByIds(FindUsersByIdsReqDTO findUsersByIdsReqDTO) {
        // 需要查询的用户 ID 集合
        List<Long> userIds = findUsersByIdsReqDTO.getIds();

        // 构建 Redis key 集合
        List<String> userInfoRedisKeys = userIds.stream()
                .map(RedisKeyConstants::buildUserInfoKey)
                .toList();

        // 先从 Redis 缓存中查, multiGet 批量查询提升性能
        List<Object> redisValues = redisTemplate.opsForValue().multiGet(userInfoRedisKeys);

        // 如果缓存中不为空
        if (CollUtil.isNotEmpty(redisValues)) {
            // 过滤掉为空的数据
            redisValues = redisValues.stream().filter(Objects::nonNull).toList();
        }

        // 反参
        List<FindUserByIdRspDTO> findUserByIdRspDTOS = Lists.newArrayList();

        // 将过滤后的缓存集合，转换为 DTO 返参实体类
        if (CollUtil.isNotEmpty(redisValues)) {
            findUserByIdRspDTOS = redisValues.stream()
                    .map(value -> JsonUtils.parseObject(String.valueOf(value), FindUserByIdRspDTO.class))
                    .collect(Collectors.toList());
        }

        // 如果被查询的用户信息，都在 Redis 缓存中, 则直接返回
        if (CollUtil.size(userIds) == CollUtil.size(findUserByIdRspDTOS)) {
            return Response.success(findUserByIdRspDTOS);
        }

        // 还有另外两种情况：一种是缓存里没有用户信息数据，还有一种是缓存里数据不全，需要从数据库中补充
        // 筛选出缓存里没有的用户数据，去查数据库
        List<Long> userIdsNeedQuery = null;

        if (CollUtil.isNotEmpty(redisValues)) {
            // 将 findUserInfoByIdRspDTOS 集合转 Map
            Map<Long, FindUserByIdRspDTO> map = findUserByIdRspDTOS.stream()
                    .collect(Collectors.toMap(FindUserByIdRspDTO::getId, p -> p));

            // 筛选出需要查 DB 的用户 ID
            userIdsNeedQuery = userIds.stream()
                    .filter(id -> Objects.isNull(map.get(id)))
                    .toList();
        }
        // 缓存中一条用户信息都没查到，则提交的用户 ID 集合都需要查数据库
        else {
            userIdsNeedQuery = userIds;
        }

        // 从数据库中批量查询
        List<UserDO> userDOS = userDOMapper.selectByIds(userIdsNeedQuery);

        List<FindUserByIdRspDTO> findUserByIdRspDTOS2 = null;

        // 若数据库查询的记录不为空
        if (CollUtil.isNotEmpty(userDOS)) {

            // 从数据库中批量查询用户语言信息
            List<UserLanguageDO> userLanguageDOS = userLanguageDOMapper.selectByUserIds(userIdsNeedQuery);

            // 提取其中的 LanguageId 查询语言信息
            List<Long> languageIds = userLanguageDOS.stream()
                    .map(UserLanguageDO::getLanguageId)
                    // 去重
                    .distinct()
                    .toList();
            List<LanguageDO> languageDOS = languageDOMapper.selectByLanguageIds(languageIds);

            // 根据语言 ID 分组
            Map<Long, LanguageDO> languageDOMap = languageDOS.stream()
                    .collect(Collectors.toMap(LanguageDO::getId, p -> p));

            // 根据用户 ID 分组
            Map<Long, List<UserLanguageDO>> userLanguageDOMap = userLanguageDOS.stream()
                    .collect(Collectors.groupingBy(UserLanguageDO::getUserId));

            // 初始化固定长度数组
            findUserByIdRspDTOS2 = Lists.newArrayListWithCapacity(10);

            // 遍历用户信息集合，构建反参数组
            for (UserDO userDO : userDOS) {
                Long userId = userDO.getId();

                // 根据 userId 获取对应用户语言信息
                List<UserLanguageDO> userLanguageDOSByUserId = userLanguageDOMap.get(userId);

                // 根据语言类型分组
                // 目的：该用户的语言信息按照类型进行分类
                // 并将 List<UserLanguageDO> 数据转化为 List<UserLanguageRspDTO> 反参类型
                Map<Integer, List<UserLanguageRspDTO>> userLanageByTypeMap = userLanguageDOSByUserId.stream()
                        .collect(
                                Collectors.groupingBy(
                                        // 根据语言类型分组
                                        UserLanguageDO::getType,
                                        // 映射为 List<UserLanguageRspDTO>
                                        Collectors.mapping(
                                                p -> {
                                                    // 获取语言信息
                                                    LanguageDO languageDO = languageDOMap.get(p.getLanguageId());
                                                    // 构建返参
                                                    return UserLanguageRspDTO.builder()
                                                            .id(languageDO.getId())
                                                            .name(languageDO.getName())
                                                            .initialsName(languageDO.getInitialsName())
                                                            .level(p.getLevel())
                                                            .build();
                                                },
                                                Collectors.toList()
                                        )
                                )
                        );

                // 构建数组返参
                FindUserByIdRspDTO findUserByIdRspDTO = FindUserByIdRspDTO.builder()
                        .id(userDO.getId())
                        .photoUrl(userDO.getPhotoUrl())
                        .displayName(userDO.getDisplayName())
                        .nativeLanguage(userLanageByTypeMap.get(LanguageTypeEnum.NATIVE.getValue()))
                        .interestLanguage(userLanageByTypeMap.get(LanguageTypeEnum.INTEREST.getValue()))
                        .build();

                findUserByIdRspDTOS2.add(findUserByIdRspDTO);
            }

            // 异步线程将用户信息同步到 Redis 中
            List<FindUserByIdRspDTO> finalFindUserByIdRspDTOS = findUserByIdRspDTOS2;
            threadPoolTaskExecutor.submit(() -> {
                // DTO 转 Map
                Map<Long, FindUserByIdRspDTO> map = finalFindUserByIdRspDTOS.stream()
                        .collect(Collectors.toMap(FindUserByIdRspDTO::getId, p -> p));

                // 执行 pipeline 操作
                redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
                    for (UserDO userDO : userDOS) {
                        Long userId = userDO.getId();

                        // 用户信息缓存 Redis Key
                        String userInfoRedisKey = RedisKeyConstants.buildUserInfoKey(userId);

                        // DTO 转字符串
                        FindUserByIdRspDTO findUserByIdRspDTO = map.get(userId);
                        String value = JsonUtils.toJsonString(findUserByIdRspDTO);

                        // 过期时间（保底1天 + 随机秒数，将缓存过期时间打散，防止同一时间大量缓存失效，导致数据库压力太大）
                        long expireSeconds = 60 * 60 * 24 + RandomUtil.randomInt(60 * 60 * 24);
                        redisTemplate.opsForValue().set(userInfoRedisKey, value, expireSeconds, TimeUnit.SECONDS);
                    }
                    return null;
                });
            });
        }

        // 合并数据
        if (CollUtil.isNotEmpty(findUserByIdRspDTOS2)) {
            findUserByIdRspDTOS.addAll(findUserByIdRspDTOS2);
        }

        return Response.success(findUserByIdRspDTOS);
    }

    /**
     * 获取用户主页信息
     *
     * @param findUserProfileReqVO
     * @return
     */
    @Override
    public Response<FindUserProfileRspVO> findUserProfile(FindUserProfileReqVO findUserProfileReqVO) {
        // 要查询的用户 ID
        Long userId = findUserProfileReqVO.getUserId();

        // 若入参用户 ID 为空，则查询当前登录用户
        if (Objects.isNull(userId)) {
            userId = LoginUserContextHolder.getUserId();
        }

        // 优先查本地缓存
        // 如果是用户本人查看自己的主页，则不走本地缓存（对本人保证实时性）
        if (!Objects.equals(userId, LoginUserContextHolder.getUserId())) {
            FindUserProfileRspVO userProfileLocalCache = PROFILE_LOCAL_CACHE.getIfPresent(userId);
            if (Objects.nonNull(userProfileLocalCache)) {
                log.info("## 用户主页信息命中本地缓存: {}", JsonUtils.toJsonString(userProfileLocalCache));
                return Response.success(userProfileLocalCache);
            }
        }

        // 优先查询 Redis 缓存
        String userProfileRedisKey = RedisKeyConstants.buildUserProfileKey(userId);

        String userProfileJson = (String) redisTemplate.opsForValue().get(userProfileRedisKey);

        if (StringUtils.isNotBlank(userProfileJson)) {
            FindUserProfileRspVO findUserProfileRspVO = JsonUtils.parseObject(userProfileJson, FindUserProfileRspVO.class);
            // 异步同步到本地缓存
            syncUserProfile2LocalCache(userId, findUserProfileRspVO);
            // 如果是博主本人查看，保证计数的实时性
            authorGetActualCountData(userId, findUserProfileRspVO);
            return Response.success(findUserProfileRspVO);
        }

        // 若 Redis 中无缓存，再查询数据库
        UserDO userDO = userDOMapper.selectById(userId);

        if (Objects.isNull(userDO)) {
            throw new BizException(ResponseCodeEnum.USER_NOT_FOUND);
        }

        // 查询用户语言信息
        List<UserLanguageDO> userLanguageDOS = userLanguageDOMapper.selectByUserId(userId);

        // 判空
        if (CollUtil.isEmpty(userLanguageDOS)) {
            throw new BizException(ResponseCodeEnum.USER_LANGUAGE_NOT_FOUND);
        }

        // 使用 stream 流转化成 languageIds 数组
        List<Long> languageIds = userLanguageDOS.stream()
                .map(UserLanguageDO::getLanguageId)
                .toList();

        // 通过 languageIds 查询出对应的语言数据
        List<LanguageDO> languageDOS = languageDOMapper.selectByLanguageIds(languageIds);

        // 构建出 languageDOMap 以 id 为 key, name 为 value
        Map<Long, LanguageDO> languageDOMap = languageDOS.stream()
                .collect(Collectors.toMap(LanguageDO::getId, value -> value));

        // 构建用户语言信息反参
        List<UserLanguageRspDTO> nativeLanguage = Lists.newArrayList();
        List<UserLanguageRspDTO> interestLanguage = Lists.newArrayList();

        // 遍历用户语言信息
        for (UserLanguageDO userLanguageDO : userLanguageDOS) {
            // 获取语言信息
            LanguageDO languageDO = languageDOMap.get(userLanguageDO.getLanguageId());
            // 构建反参
            UserLanguageRspDTO userLanguageRspDTO = UserLanguageRspDTO.builder()
                    .id(userLanguageDO.getLanguageId())
                    .name(languageDO.getName())
                    .initialsName(languageDO.getInitialsName())
                    .level(userLanguageDO.getLevel())
                    .build();

            LanguageTypeEnum languageTypeEnum = LanguageTypeEnum.valueOf(userLanguageDO.getType());

            // 判断加入的语言数组
            switch (languageTypeEnum) {
                // 添加到母语中
                case NATIVE -> nativeLanguage.add(userLanguageRspDTO);
                // 添加到感兴趣语言中
                case INTEREST -> interestLanguage.add(userLanguageRspDTO);
            }

        }

        // 构建返参 VO
        FindUserProfileRspVO findUserProfileRspVO = FindUserProfileRspVO.builder()
                .userId(userDO.getId())
                .photoUrl(userDO.getPhotoUrl())
                .displayName(userDO.getDisplayName())
                .synopsis(userDO.getSynopsis())
                .nativeLanguage(nativeLanguage)
                .interestLanguage(interestLanguage)
                .build();

        // RPC: Feign 调用计数服务
        // 发布的问题数，获得的点赞数、回答数、获得最佳回答数、浏览数
        rpcCountServiceAndSetData(userId, findUserProfileRspVO);

        // 异步同步到 Redis 中
        syncUserProfile2Redis(userProfileRedisKey, findUserProfileRspVO);
        // 异步同步到本地缓存
        syncUserProfile2LocalCache(userId, findUserProfileRspVO);

        return Response.success(findUserProfileRspVO);
    }

    /**
     * Feign 调用计数服务, 并设置计数数据
     *
     * @param userId
     * @param findUserProfileRspVO
     */
    private void rpcCountServiceAndSetData(Long userId, FindUserProfileRspVO findUserProfileRspVO) {
        FindUserCountsByIdRspDTO findUserCountsByIdRspDTO = countRpcService.findUserCountById(userId);

        if (Objects.nonNull(findUserCountsByIdRspDTO)) {
            Long quizTotal = findUserCountsByIdRspDTO.getQuizTotal();
            Long commentTotal = findUserCountsByIdRspDTO.getCommentTotal();
            Long likeTotal = findUserCountsByIdRspDTO.getLikeTotal();
            Long browseTotal = findUserCountsByIdRspDTO.getBrowseTotal();
            Long bestCommentTotal = findUserCountsByIdRspDTO.getBestCommentTotal();

            findUserProfileRspVO.setQuizTotal(NumberUtils.formatNumberString(quizTotal));
            findUserProfileRspVO.setCommentTotal(NumberUtils.formatNumberString(commentTotal));
            findUserProfileRspVO.setLikeTotal(NumberUtils.formatNumberString(likeTotal));
            findUserProfileRspVO.setBrowseTotal(NumberUtils.formatNumberString(browseTotal));
            findUserProfileRspVO.setBestCommentTotal(NumberUtils.formatNumberString(bestCommentTotal));
        }
    }

    /**
     * 作者本人获取真实的计数数据（保证实时性）
     *
     * @param userId
     * @param findUserProfileRspVO
     */
    private void authorGetActualCountData(Long userId, FindUserProfileRspVO findUserProfileRspVO) {
        if (Objects.equals(userId, LoginUserContextHolder.getUserId())) {
            rpcCountServiceAndSetData(userId, findUserProfileRspVO);
        }
    }

    /**
     * 异步同步到 Redis 中
     *
     * @param userProfileRedisKey
     * @param findUserProfileRspVO
     */
    private void syncUserProfile2Redis(String userProfileRedisKey, FindUserProfileRspVO findUserProfileRspVO) {
        threadPoolTaskExecutor.submit(() -> {
            // 设置随机过期时间 (2小时以内)
            long expireTime = 60 * 60 + RandomUtil.randomInt(60 * 60);

            // 将 VO 转为 Json 字符串写入到 Redis 中
            redisTemplate.opsForValue().set(userProfileRedisKey, JsonUtils.toJsonString(findUserProfileRspVO), expireTime, TimeUnit.SECONDS);
        });
    }

    /**
     * 异步同步到本地缓存
     *
     * @param userId
     * @param findUserProfileRspVO
     */
    private void syncUserProfile2LocalCache(Long userId, FindUserProfileRspVO findUserProfileRspVO) {
        threadPoolTaskExecutor.submit(() -> {
            PROFILE_LOCAL_CACHE.put(userId, findUserProfileRspVO);
        });
    }

    /**
     * 删除 Redis 中的用户缓存
     *
     * @param userId
     */
    private void deleteUserRedisCache(Long userId) {
        // 构建 Redis Key
        String userInfoRedisKey = RedisKeyConstants.buildUserInfoKey(userId);
        String userProfileRedisKey = RedisKeyConstants.buildUserProfileKey(userId);

        // 批量删除
        redisTemplate.delete(Arrays.asList(userInfoRedisKey, userProfileRedisKey));
    }

    /**
     * 异步发送延时消息
     *
     * @param userId
     */
    private void sendDelayDeleteUserRedisCacheMQ(Long userId) {
        Message<String> message = MessageBuilder.withPayload(String.valueOf(userId)).build();

        rocketMQTemplate.asyncSend(MQConstants.TOPIC_DELAY_DELETE_USER_REDIS_CACHE, message,
                new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        log.info("## 延时删除 Redis 用户缓存消息发送成功...");
                    }

                    @Override
                    public void onException(Throwable e) {
                        log.error("## 延时删除 Redis 用户缓存消息发送失败...", e);
                    }
                },
                // 超时时间
                3000,
                // 延迟级别，1 表示延时 1s
                1);
    }
}
